package org.atlantis.ymer;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;

import org.atlantis.ymer.EventHelper.EventDispatcherHolder;
import org.atlantis.ymer.command.Command;
import org.atlantis.ymer.command.CommandType;
import org.atlantis.ymer.command.ConnectCommand;
import org.atlantis.ymer.command.DisconnectCommand;
import org.atlantis.ymer.command.PubCommand;
import org.atlantis.ymer.common.IdGenerator;
import org.atlantis.ymer.event.ClientEventEmitter;
import org.atlantis.ymer.event.ServerEventEmitter;
import org.jboss.netty.util.HashedWheelTimer;
import org.jboss.netty.util.Timeout;
import org.jboss.netty.util.Timer;
import org.jboss.netty.util.TimerTask;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

class WebSessionFactoryImpl implements WebSessionFactory {

	private static final Logger log = LoggerFactory.getLogger(WebSessionFactoryImpl.class);

	private ConcurrentMap<String, WebSession> sessions = new ConcurrentHashMap<String, WebSession>(8192);
	private Timer timer = new HashedWheelTimer(500, TimeUnit.MILLISECONDS);

	private ServerEventEmitterImpl serverDispatcher = new ServerEventEmitterImpl();

	// WebSession's options
	private long heartbeatInterval;
	private long timeoutInterval;

	WebSessionFactoryImpl(YmerOptions options) {
		this.heartbeatInterval = options.getHeartbeatInterval();
		this.timeoutInterval = options.getTimeoutInterval();
	}

	ServerEventEmitter getServerDispatcher() {
		return serverDispatcher;
	}

	@Override
	public WebSession create() {
		WebSession webSession = new WebSessionImpl(IdGenerator.generate(), heartbeatInterval, timeoutInterval);
		sessions.put(webSession.getId(), webSession);
		return webSession;
	}

	@Override
	public WebSession get(String id) {
		return sessions.get(id);
	}

	@Override
	public void dispose() {
		sessions.clear();
		timer.stop();
		serverDispatcher.dispose();
	}

	private class WebSessionImpl extends EventDispatcherHolder implements WebSession {

		private final String id;
		private DataForwarder forwarder;
		private volatile boolean connected = false;
		private long heartbeatInterval;
		private long timeoutInterval;
		private Timeout heartbeatTask;
		private Timeout timeoutTask;

		WebSessionImpl(String id, long heartbeatInterval, long timeoutInterval) {
			this.id = id;
			this.heartbeatInterval = heartbeatInterval;
			this.timeoutInterval = timeoutInterval;
		}

		@Override
		public String getId() {
			return id;
		}

		@Override
		public boolean isConnected() {
			return connected;
		}

		@Override
		public void onData(Command command) {
			CommandType type = command.getType();
			switch (type) {
			case CONNECT:
				onConnect((ConnectCommand) command);
				return;
			case DISCONNECT:
				onDisconnect();
				return;
			case PONG:
				onPong();
				return;
			case PUB:
				onPub((PubCommand) command);
				return;
			default:
				return;
			}
		}

		private void startHeartbeatTask() {
			cancelHeartbeatTask();
			if (heartbeatInterval <= 0) return;
			heartbeatTask = timer.newTimeout(new TimerTask() {
				@Override
				public void run(Timeout timeout) throws Exception {
					sendPing();
				}
			}, heartbeatInterval, TimeUnit.MILLISECONDS);
		}

		private void startTimeoutTask() {
			cancelTimeoutTask();
			if (timeoutInterval <= 0) return;
			timeoutTask = timer.newTimeout(new TimerTask() {
				@Override
				public void run(Timeout timeout) throws Exception {
					close(DisconnectReason.TIMEOUT_DISCONNECT);
				}
			}, timeoutInterval, TimeUnit.MILLISECONDS);
		}

		private void cancelHeartbeatTask() {
			if (heartbeatTask == null) return;
			heartbeatTask.cancel();
			heartbeatTask = null;
		}

		private void cancelTimeoutTask() {
			if (timeoutTask == null) return;
			timeoutTask.cancel();
			timeoutTask = null;
		}

		@Override
		public void close(DisconnectReason reason) {
			if (!connected) return;
			connected = false;
			cancelHeartbeatTask();
			cancelTimeoutTask();
			sessions.remove(id);
			if (reason.isNoticeClient()) forwarder.sendDisconnect(new DisconnectCommand(reason));
			try {
				dispatcher.dispatch(ClientEventEmitter.AppTopic.DISCONNECT, reason);
			} catch (Throwable cause) {
				log.error("disconnect topic error.", cause);
			}
		}

		private void onConnect(ConnectCommand command) {
			forwarder = command.getForwarder();
			// send connect ack to client
			forwarder.sendConnect();
			startHeartbeatTask();
			connected = true;
			try {
				serverDispatcher.dispatch(ServerEventEmitter.AppTopic.CONNECT, this);
			} catch (Throwable cause) {
				log.error("connect topic error.", cause);
				close(DisconnectReason.createException(cause));
			}
		}

		private void onDisconnect() {
			close(DisconnectReason.NORMAL_DISCONNECT);
		}

		private void sendPing() {
			forwarder.sendPing();
			startTimeoutTask();
		}

		private void onPong() {
			cancelTimeoutTask();
			startHeartbeatTask();
		}

		private void onPub(PubCommand command) {
			try {
				dispatcher.dispatch(command.getTopic(), command);
			} catch (Throwable cause) {
				log.error("onMessage() error. Topic: " + command.getTopic(), cause);
				close(DisconnectReason.createException(cause));
			}
		}

		@Override
		public DataForwarder getBindedForwarder() {
			return forwarder;
		}
	}
}
